<?php

namespace Common\Mqtt;

use Common\BaseTable;
use Common\Event\SendToMqtt;
use Common\Event\SendToTcp;
use Common\Helpers\DebugHelper;
use Common\Mqtt\Utility\MqttUtility;
use Hyperf\Context\ApplicationContext;
use Hyperf\Contract\ConfigInterface;
use Hyperf\Contract\OnReceiveInterface;
use Hyperf\Di\Annotation\Inject;
use Psr\Container\ContainerExceptionInterface;
use Psr\Container\NotFoundExceptionInterface;
use Psr\EventDispatcher\EventDispatcherInterface;
use Psr\SimpleCache\CacheInterface;
use Simps\MQTT\Client;
use Simps\MQTT\Config\ClientConfig;
use Simps\MQTT\Protocol\ProtocolInterface;
use Simps\MQTT\Protocol\Types;
use Simps\MQTT\Protocol\V3;
use Simps\MQTT\Protocol\V5;
use Simps\MQTT\Tools\UnPackTool;

/**
 * mqtt生命周期类
 */
class MqttServer implements OnReceiveInterface
{
    /**
     * @Inject
     * @var EventDispatcherInterface
     */
    #[Inject]
    protected EventDispatcherInterface $eventDispatcher;



    /**
     * @Inject
     * @var CacheInterface
     */
    #[Inject]
    protected CacheInterface $cache;

    public function onConnect(): void
    {
        echo 'mqtt-onConnect'.PHP_EOL;
        BaseTable::getInstance();
    }

    public function onClose(): void
    {
        echo 'mqtt-onClose'.PHP_EOL;
    }

    /**
     *
     */
    public function onReceive($server, int $fd, int $reactorId, string $data): void
    {
        echo 'mqtt-onReceive'."[fd]$fd".PHP_EOL;
        $type = UnPackTool::getType($data);
        $class = V5::class;

        if ($type === Types::CONNECT) {
            $level = UnPackTool::getLevel($data);
            $class = $level === ProtocolInterface::MQTT_PROTOCOL_LEVEL_5_0 ? V5::class : V3::class;
        }
        try {
            // debug
//        Common::printf($data);
            $data = $class::unpack($data);
            DebugHelper::consoleWrite('mqtt-onReceive',$data);

            if (isset($data['type'])) {
                switch ($data['type']) {
                    case Types::CONNECT:
                        // Check protocol_name
                        if ($data['protocol_name'] != 'MQTT') {
                            DebugHelper::consoleWrite('mqtt-onReceive-关闭',[$fd]);
                            $server->close($fd);
                            break;
                        }
                        // Check connection information, etc.
                        $server->send(
                            $fd,
                            $class::pack(
                                [
                                    'type' => Types::CONNACK,
                                    'code' => 0,
                                    'session_present' => 0,
                                    'properties' => [
                                        'maximum_packet_size' => 1048576,
                                        'retain_available' => true,
                                        'shared_subscription_available' => true,
                                        'subscription_identifier_available' => true,
                                        'topic_alias_maximum' => 65535, //0
                                        'wildcard_subscription_available' => true,
                                    ],
                                ]
                            )
                        );
                        break;
                    case Types::PINGREQ:
                        $server->send($fd, $class::pack(['type' => Types::PINGRESP]));
                        break;
                    case Types::DISCONNECT:
                        if ($server->exist($fd)) {
                            $server->close($fd);
                        }
                        break;
                    case Types::PUBLISH:
                        $MqttUtility = new MqttUtility();
                        // Send to subscribers 遍历所有连接，发送消息，已确认其他连接订阅的主题回复
                        foreach ($server->connections as $sub_fd) {
//                            $a =  $server->getClientInfo($sub_fd);
//                            $b = $server->getClientInfo($fd);

                            // 检查是否为发送方
                            if ($sub_fd == $fd) {
                                continue;
                            }

                            // 检查是否为WebSocket连接
                            if ($MqttUtility->isWebSocketConnection($server,$sub_fd)) {
                                DebugHelper::consoleWrite('mqtt-onReceive-是websocket连接，跳过',[$sub_fd]);
                                continue;
                            }

                            $server->send(
                                $sub_fd,
                                $class::pack(
                                    [
                                        'type' => $data['type'],
                                        'topic' => $data['topic'],
                                        'message' => $data['message'],
                                        'dup' => $data['dup'],
                                        'qos' => $data['qos'],
                                        'retain' => $data['retain'],
                                        'message_id' => $data['message_id'] ?? 0,
                                    ]
                                )
                            );
                        }

                        if ($data['qos'] === 1) {
                            $server->send(
                                $fd,
                                $class::pack(
                                    [
                                        'type' => Types::PUBACK,
                                        'message_id' => $data['message_id'] ?? 0,
                                    ]
                                )
                            );
                        }

                        break;
                    case Types::SUBSCRIBE:
                        $payload = [];
                        DebugHelper::consoleWrite('mqtt-onReceive-SUBSCRIBE订阅消息',$data);

                        foreach ($data['topics'] as $k => $option) {
                            $qos = $option['qos'];
                            if (is_numeric($qos) && $qos < 3) {
                                $payload[] = $qos;
                            } else {
                                $payload[] = \Simps\MQTT\Hex\ReasonCode::QOS_NOT_SUPPORTED;
                            }
                        }
                        $server->send(
                            $fd,
                            $class::pack(
                                [
                                    'type' => Types::SUBACK,
                                    'message_id' => $data['message_id'] ?? 0,
                                    'codes' => $payload,
                                ]
                            )
                        );
                        break;
                    case Types::UNSUBSCRIBE:
                        $payload = [];
                        foreach ($data['topics'] as $k => $qos) {
                            if (is_numeric($qos) && $qos < 3) {
                                $payload[] = $qos;
                            } else {
                                $payload[] = 0x80;
                            }
                        }
                        $server->send(
                            $fd,
                            $class::pack(
                                [
                                    'type' => Types::UNSUBACK,
                                    'message_id' => $data['message_id'] ?? 0,
                                    'codes' => $payload,
                                ]
                            )
                        );
                        break;
                }
            } else {
                DebugHelper::consoleWrite('mqtt-onReceive-clide',[$fd]);

                $server->close($fd);
            }
        } catch (\Throwable $e) {
            echo "\033[0;31mError: {$e->getMessage()}\033[0m\r\n";
            $server->close($fd);
        }

    }
}
